package com.yiguo.realtime.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.google.gson.Gson
import com.yiguo.gmall.Constansts
import com.yiguo.realtime.utils.{MyKafkaUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.InputDStream
import redis.clients.jedis.Jedis

object UserInfoApp extends BaseApp {
  override var appName: String = "UserInfoApp"
  override var duration: Int = 10

  def main(args: Array[String]): Unit = {
    run{
      //消费kafka  获取ds
      val ds: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(Constansts.GMALL_USER_INFO, streamingContext)
      //把数据直接写入redis
      ds.foreachRDD(rdd=>{

        rdd.foreachPartition(partition=>{
          //获取redis连接
          val jedis: Jedis = RedisUtil.getJedisClient()
          partition.foreach(record=>{
            val key: String = JSON.parseObject(record.value()).getString("id")
            jedis.set("user_id"+key,record.value())
          })
          //关闭连接
          jedis.close()
        })

      })

    println("拉取数据中...请稍后")
    }
  }
}
